package labox.innovation.db.core;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.LinkedList;
import java.util.Queue;
import javax.sql.ConnectionEvent;
import javax.sql.ConnectionEventListener;
import javax.sql.ConnectionPoolDataSource;
import javax.sql.PooledConnection;

import labox.innovation.config.Config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public final class MiniConnectionPoolManager {

	public static final Logger _log = LoggerFactory.getLogger(MiniConnectionPoolManager.class.getName());
	private static ConnectionPoolDataSource _dataSource;
	private static int _maxConnections;
	private static int _timeout;

	// Semaphore ͨ���������ƿ��Է���ĳЩ��Դ��������߼��ģ����߳���Ŀ��

	private static Semaphore semaphore;
	private final static Queue<PooledConnection> recycledConnections = new LinkedList<PooledConnection>();
	private static int activeConnections;
	private final static PoolConnectionEventListener poolConnectionEventListener = new PoolConnectionEventListener();
	private static boolean isDisposed;

	/**
	 * Thrown in {@link #getConnection()} when no free connection becomes
	 * available within <code>timeout</code> seconds.
	 */
	public static class TimeoutException extends RuntimeException {
		private static final long serialVersionUID = 1;

		public TimeoutException() {
			super("Timeout while waiting for a free database connection.");
		}
	}

	/**
	 * Constructs a MiniConnectionPoolManager object with a timeout of 60
	 * seconds.
	 * 
	 * @param dataSource
	 *            the data source for the connections.
	 * @param maxConnections
	 *            the maximum number of connections.
	 */
	public static final void getMiniConnectionPoolManager(ConnectionPoolDataSource dataSource, int maxConnections) {
		getMiniConnectionPoolManager(dataSource, maxConnections, 60);
	}

	/**
	 * Constructs a MiniConnectionPoolManager object.
	 * 
	 * @param _dataSource
	 *            the data source for the connections.
	 * @param maxConnections
	 *            the maximum number of connections.
	 * @param timeout
	 *            the maximum time in seconds to wait for a free connection.
	 */
	public static final void getMiniConnectionPoolManager(ConnectionPoolDataSource dataSource, int maxConnections, int timeout) {
		_dataSource = dataSource;
		_maxConnections = maxConnections;
		_timeout = timeout;
		if (maxConnections < 1)
			throw new IllegalArgumentException("Invalid maxConnections value.");
		semaphore = new Semaphore(maxConnections, true);

	}

	/**
	 * Closes all unused pooled connections.
	 */
	public static final synchronized void dispose() throws SQLException {
		if (isDisposed)
			return;
		isDisposed = true;
		SQLException e = null;
		while (!recycledConnections.isEmpty()) {
			PooledConnection pconn = recycledConnections.remove();
			try {
				pconn.close();
			} catch (SQLException e2) {
				_log.error("Error MiniConnectionPoolManager dispose", Config.SERVER_ID, e);
				if (e == null)
					e = e2;
			}
		}
		if (e != null)
			throw e;
	}

	/**
	 * Retrieves//ȡ�� a connection from the connection pool. If
	 * <code>maxConnections</code> connections are already in use, the method
	 * waits until a connection becomes available or <code>timeout</code>
	 * seconds elapsed.//���� When the application is finished using the
	 * connection, it must close it in order to return it to the pool.
	 * 
	 * @return a new Connection object.
	 * @throws TimeoutException
	 *             when no connection becomes available within
	 *             <code>timeout</code> seconds.
	 */
	public final static Connection getConnection() throws SQLException {
		// This routine is unsynchronized, because semaphore.tryAcquire() may
		// block.
		synchronized (MiniConnectionPoolManager.class) {
			if (isDisposed)
				throw new IllegalStateException("Connection pool has been disposed.");
		}
		try {
			if (!semaphore.tryAcquire(_timeout, TimeUnit.SECONDS))
				throw new TimeoutException();
		} catch (InterruptedException e) {
			throw new RuntimeException("Interrupted while waiting for a database connection.", e);
		}
		boolean ok = false;
		try {
			Connection conn = getConnection2();
			ok = true;
			return conn;
		} finally {
			if (!ok)
				semaphore.release();
		}
	}

	private synchronized static final Connection getConnection2() throws SQLException {
		if (isDisposed)
			throw new IllegalStateException("Connection pool has been disposed."); // test
		// again
		// with
		// lock
		PooledConnection pconn;
		if (!recycledConnections.isEmpty()) {
			pconn = recycledConnections.remove();
		} else {
			pconn = _dataSource.getPooledConnection();
		}
		Connection conn = pconn.getConnection();
		activeConnections++;
		pconn.addConnectionEventListener(poolConnectionEventListener);
		assertInnerState();
		return conn;
	}

	private synchronized static final void recycleConnection(PooledConnection pconn) {
		if (isDisposed) {
			disposeConnection(pconn);
			return;
		}
		if (activeConnections <= 0)
			throw new AssertionError();
		activeConnections--;
		semaphore.release();
		recycledConnections.add(pconn);
		assertInnerState();
	}

	private synchronized static final void disposeConnection(PooledConnection pconn) {
		if (activeConnections <= 0)
			throw new AssertionError();
		activeConnections--;
		semaphore.release();
		closeConnectionNoEx(pconn);
		assertInnerState();
	}

	private static final void closeConnectionNoEx(PooledConnection pconn) {
		try {
			pconn.close();
		} catch (SQLException e) {
			_log.error("Error while closing database connection: " + e.toString());
		}
	}

	private static final void assertInnerState() {
		if (activeConnections < 0)
			throw new AssertionError();
		if (activeConnections + recycledConnections.size() > _maxConnections)
			throw new AssertionError();
		if (activeConnections + semaphore.availablePermits() > _maxConnections)
			throw new AssertionError();
	}

	private static class PoolConnectionEventListener implements ConnectionEventListener {
		public void connectionClosed(ConnectionEvent event) {
			PooledConnection pconn = (PooledConnection) event.getSource();
			pconn.removeConnectionEventListener(this);
			recycleConnection(pconn);
		}

		public void connectionErrorOccurred(ConnectionEvent event) {
			PooledConnection pconn = (PooledConnection) event.getSource();
			pconn.removeConnectionEventListener(this);
			disposeConnection(pconn);
		}
	}

	/**
	 * Returns the number of active (open) connections of this pool. This is the
	 * number of <code>Connection</code> objects that have been issued by
	 * {@link #getConnection()} for which <code>Connection.close()</code> has
	 * not yet been called.
	 * 
	 * @return the number of active connections.
	 **/
	public synchronized int getActiveConnections() {
		return activeConnections;
	}

} // end class MiniConnectionPoolManager
